package syR

import (
	cp "db2s/componentInit"
	mq "db2s/topic-mq"
	"reflect"
	"strings"
)

func mqListenersTableInit(t, p int) map[Topic]*cp.MQTopicListeners {
	return map[Topic]*cp.MQTopicListeners{
		//sync rows同步全量数据
		syncIsBaseQuery:          cp.NewMQTopicListeners(cp.Topic(syncIsBaseQuery), t, singleQueryMessageQueueLength, syncIsBaseQueryListeners, 5, 10),
		syncIndexConcurrentQuery: cp.NewMQTopicListeners(cp.Topic(syncIndexConcurrentQuery), t*p, 1000, syncIsSecondQueryListeners, 5, 20),
		syncQueryTableSql:        cp.NewMQTopicListeners(cp.Topic(syncQueryTableSql), t*p, 1000, querySqlListeners, 1, 30),
		syncQueryTableData:       cp.NewMQTopicListeners(cp.Topic(syncQueryTableData), t*p*2, 1000, queryDataListeners, 5, 41),
		syncGeneralInsert:        cp.NewMQTopicListeners(cp.Topic(syncGeneralInsert), t*p*2, 1000, generalInsertListeners, 5, 42),
		syncWriteInsert:          cp.NewMQTopicListeners(cp.Topic(syncWriteInsert), t*p*2, 1000, writeInsertListeners, 5, 43),
	}
}
func syncDataMQListenInitFunc(t, p int) map[cp.CheckMode][]*cp.MQTopicListeners {
	MQListenersTable := mqListenersTableInit(t, p)
	return map[cp.CheckMode][]*cp.MQTopicListeners{
		"sync": {
			MQListenersTable[syncIsBaseQuery],
			MQListenersTable[syncIndexConcurrentQuery],
			MQListenersTable[syncQueryTableSql],
			MQListenersTable[syncQueryTableData],
			MQListenersTable[syncGeneralInsert],
			MQListenersTable[syncWriteInsert],
		},
	}
}
func topicCreateInit(checkMod string, t, p int) (m map[cp.CheckMode][]*cp.MQTopicListeners) {
	m = make(map[cp.CheckMode][]*cp.MQTopicListeners)
	if funcName, ok := MqListenerSInitFunc[checkMod]; ok {
		if funcName.Kind() == reflect.Func {
			m1 := funcName.Call([]reflect.Value{
				reflect.ValueOf(t),
				reflect.ValueOf(p),
			})
			for i := 0; i < len(m1); i++ {
				m = m1[i].Interface().(map[cp.CheckMode][]*cp.MQTopicListeners)
			}
		}
	}
	return m
}
func GetCheckToTopicMQListeners() map[cp.CheckMode][]*cp.MQTopicListeners {
	return topicCreateInit(GlobalPConfigs.rules.CheckMode, GlobalPConfigs.rules.Mtc, GlobalPConfigs.rules.Mcc)
}

func (context *SchedulePlanContext) topicEndAdd(index CheckMode, topic Topic) {
	context.AutoCloseMqComponentTable[cp.CheckMode(index)].TopicEndAdd(cp.Topic(topic))
}
func (context *SchedulePlanContext) topicEndMonit(index CheckMode, topic Topic) int64 {
	return context.AutoCloseMqComponentTable[cp.CheckMode(index)].TopicEndMonit(cp.Topic(topic))
}
func (context *SchedulePlanContext) topicEndDelete(index CheckMode, topic Topic) {
	context.AutoCloseMqComponentTable[cp.CheckMode(index)].TopicEndDelete(cp.Topic(topic))
}

func (context *SchedulePlanContext) topicEndDeleteNumber(index CheckMode, topic Topic, n int64) {
	context.AutoCloseMqComponentTable[cp.CheckMode(index)].TopicEndDeleteNumber(cp.Topic(topic), n)
}

func sendMsg(check CheckMode, topic Topic, tableName string, msg map[string]any) bool {
	var outFay string
	PlanContext.topicEndAdd(check, topic)
	if err := producer.Send(mq.Message{
		Topic:      string(topic),
		Body:       []byte(tableName),
		Properties: msg,
	}); err != nil {
		return false
	}
	if v, ok := msg["outFay"]; ok {
		outFay = v.(string)
	}
	if strings.EqualFold(outFay, "bar") {
		PlanContext.taskBarEndTopicAdd(msg)
	}
	return true
}
func deleteTopic(topicMsg map[string]any) {
	var (
		dTopic Topic
		outFay = "bar"
	)
	if v, ok := topicMsg["topic"]; ok {
		dTopic = v.(Topic)
	}
	if v, ok := topicMsg["outFay"]; ok {
		outFay = v.(string)
	}
	PlanContext.topicEndDelete(getCheckMod(), dTopic)
	if strings.EqualFold(outFay, "bar") {
		PlanContext.TaskBarEndTopicDel(topicMsg)
	}
}
